package utils;


import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import scala.math.BigDecimal;

import java.lang.reflect.Field;
import java.math.BigInteger;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @auther: wuyahan
 * @Date: 2021/4/13 13:06
 * @Description:
 */
public class FlinkDataType {

    public static Map<String, Object> dataType = new HashMap();

    static {
        dataType.put("CHAR", "CHAR");
        dataType.put("VARCHAR", "VARCHAR");
        dataType.put("STRING", "STRING");
        dataType.put("BOOLEAN", "BOOLEAN");
        dataType.put("BYTES", "BYTES");
        dataType.put("DECIMAL", "DECIMAL");
        dataType.put("TINYINT", "TINYINT");
        dataType.put("SMALLINT", "SMALLINT");
        dataType.put("INTEGER", "INTEGER");
        dataType.put("BIGINT", "BIGINT");
        dataType.put("FLOAT", "FLOAT");
        dataType.put("DOUBLE", "DOUBLE");
        dataType.put("DATE", "DATE");
        dataType.put("TIME", "TIME");
        dataType.put("TIMESTAMP", "TIMESTAMP");
        dataType.put("INTERVAL", "INTERVAL");
        dataType.put("ARRAY", "ARRAY");
        dataType.put("MULTISET", "MULTISET");
        dataType.put("MAP", "MAP");
        dataType.put("ROW", "ROW");
        dataType.put("RAW", "RAW");
    }


    public static TypeInformation getTypeInformation(String dataType) {
        TypeInformation typeInformation;
        dataType = dataType.toUpperCase();
        switch (dataType) {
            case "CHAR":
                return TypeInformation.of(Character.class);
            case "STRING":
                return TypeInformation.of(String.class);
            case "VARCHAR":
                return TypeInformation.of(String.class);
            case "BOOLEAN":
                return TypeInformation.of(Boolean.class);
            case "BYTES":
                return TypeInformation.of(Byte.class);
            case "DECIMAL":
                return TypeInformation.of(BigDecimal.class);
            case "TINYINT":
                return TypeInformation.of(Long.class);
            case "SMALLINT":
                return TypeInformation.of(Integer.class);
            case "INTEGER":
                return TypeInformation.of(Integer.class);
            case "BIGINT":
                return TypeInformation.of(Long.class);
            case "FLOAT":
                return TypeInformation.of(Float.class);
            case "DOUBLE":
                return TypeInformation.of(Double.class);
            case "DATE":
                return TypeInformation.of(Date.class);
            case "DATETIME":
                return TypeInformation.of(Time.class);
            case "TIME":
                return TypeInformation.of(Time.class);
            case "TIMESTAMP":
                return Types.SQL_TIMESTAMP;
//            case "INTERVAL":
//                return Types.In;
//            case "ARRAY":
//                return Types.SQL_TIMESTAMP;
            default:
                return TypeInformation.of(String.class);
        }


    }


    //TODO 待优化，可以提取到相关工具类或者其他优化方式

    /**
     * 数据从 String 转换成对应的类型
     *
     * @param fieldType
     * @param value
     * @return
     * @throws Exception
     */
    public static Object transformValue(Class fieldType, String value) throws Exception {

        if (fieldType == String.class) {
            return String.valueOf(value);
        } else if (fieldType == Boolean.class || fieldType == boolean.class) {
            return Boolean.valueOf(value);
        } else if (fieldType == Byte.class || fieldType == byte.class) {
            return Byte.valueOf(value);
        } else if (fieldType == Double.class || fieldType == double.class) {
            return Double.valueOf(value);
        } else if (fieldType == Float.class || fieldType == float.class) {
            return Float.valueOf(value);
        } else if (fieldType == Integer.class || fieldType == int.class) {
            return Integer.valueOf(value);
        } else if (fieldType == Long.class || fieldType == long.class) {
            return Long.valueOf(value);
        } else if (fieldType == Short.class || fieldType == short.class) {
            return Short.valueOf(value);
        } else if (fieldType == Character.class || fieldType == char.class) {
            return value.charAt(0);
        } else if (fieldType == java.math.BigDecimal.class) {
            return new java.math.BigDecimal(value);
        } else if (fieldType == BigInteger.class) {
            return new BigInteger(value);
        } else if (fieldType == java.util.Date.class) {
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            //特殊处理 例：2018-05-23T10:36:48.525000
            value = value.replace("T", " ");
            return sdf.parse(value);
        } else if (fieldType == List.class) {
            return Arrays.asList(value.split(","));
        } else if (fieldType == Set.class) {
            return new HashSet<>(Arrays.asList(value.split(",")));
        } else if (fieldType.isEnum()) {
            Class<?> cl = Class.forName(fieldType.getName());
            Field field = cl.getDeclaredField(value);
            return field.get(cl);
        } else if (fieldType == Pattern.class) {
            return Pattern.compile(value);
        } else if (fieldType == Timestamp.class) {
            value = value.replace("T", " ");
            return Timestamp.valueOf(value);
        } else {
            return value;
        }
    }


//    private static CsvRowDeserializationSchema.RuntimeConverter createRuntimeConverter(
//            TypeInformation<?> info, boolean ignoreParseErrors) {
//        if (info.equals(Types.VOID)) {
//            return (node) -> null;
//        } else if (info.equals(Types.STRING)) {
//            return JsonNode::asText;
//        } else if (info.equals(Types.BOOLEAN)) {
//            return (node) -> Boolean.valueOf(node.asText().trim());
//        } else if (info.equals(Types.BYTE)) {
//            return (node) -> Byte.valueOf(node.asText().trim());
//        } else if (info.equals(Types.SHORT)) {
//            return (node) -> Short.valueOf(node.asText().trim());
//        } else if (info.equals(Types.INT)) {
//            return (node) -> Integer.valueOf(node.asText().trim());
//        } else if (info.equals(Types.LONG)) {
//            return (node) -> Long.valueOf(node.asText().trim());
//        } else if (info.equals(Types.FLOAT)) {
//            return (node) -> Float.valueOf(node.asText().trim());
//        } else if (info.equals(Types.DOUBLE)) {
//            return (node) -> Double.valueOf(node.asText().trim());
//        } else if (info.equals(Types.BIG_DEC)) {
//            return (node) -> new java.math.BigDecimal(node.asText().trim());
//        } else if (info.equals(Types.BIG_INT)) {
//            return (node) -> new BigInteger(node.asText().trim());
//        } else if (info.equals(Types.SQL_DATE)) {
//            return (node) -> Date.valueOf(node.asText());
//        } else if (info.equals(Types.SQL_TIME)) {
//            return (node) -> Time.valueOf(node.asText());
//        } else if (info.equals(Types.SQL_TIMESTAMP)) {
//            return (node) -> Timestamp.valueOf(node.asText());
//        } else if (info.equals(Types.LOCAL_DATE)) {
//            return (node) -> Date.valueOf(node.asText()).toLocalDate();
//        } else if (info.equals(Types.LOCAL_TIME)) {
//            return (node) -> Time.valueOf(node.asText()).toLocalTime();
//        } else if (info.equals(Types.LOCAL_DATE_TIME)) {
//            return (node) -> Timestamp.valueOf(node.asText()).toLocalDateTime();
//        } else if (info instanceof RowTypeInfo) {
//            final RowTypeInfo rowTypeInfo = (RowTypeInfo) info;
//            return createRowRuntimeConverter(rowTypeInfo, ignoreParseErrors, false);
//        } else if (info instanceof BasicArrayTypeInfo) {
//            return createObjectArrayRuntimeConverter(
//                    ((BasicArrayTypeInfo<?, ?>) info).getComponentInfo(), ignoreParseErrors);
//        } else if (info instanceof ObjectArrayTypeInfo) {
//            return createObjectArrayRuntimeConverter(
//                    ((ObjectArrayTypeInfo<?, ?>) info).getComponentInfo(), ignoreParseErrors);
//        } else if (info instanceof PrimitiveArrayTypeInfo
//                && ((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
//            return createByteArrayRuntimeConverter(ignoreParseErrors);
//        } else {
//            throw new RuntimeException("Unsupported type information '" + info + "'.");
//        }
//    }
}
